package com.jzsec.tzdslog;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.OutputStreamWriter;

public class SinkToTextFile extends RichSinkFunction<String> {
	private String filepath;
	private String filename;
	private FileOutputStream fos;
	private OutputStreamWriter osw;
	private BufferedWriter bufferedWriter;

	public SinkToTextFile(String filepath, String filename) {
		this.filepath = filepath;
		this.filename = filename;
		File dir = new File(filepath);
		if (!dir.isDirectory()) {
			dir.mkdir();
		}
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		File file = new File(filepath);
		if (!file.isDirectory()) {
			file.mkdir();
		}
		fos = new FileOutputStream(new File(filepath + "/" + filename), true);
		osw = new OutputStreamWriter(fos);
		bufferedWriter = new BufferedWriter(osw);
	}

	@Override
	public void close() throws Exception {
		if (bufferedWriter != null) bufferedWriter.close();
		if (osw != null) osw.close();
		if (fos != null) fos.close();
		super.close();
	}

	@Override
	public void invoke(String value, Context context) throws Exception {
		bufferedWriter.write(value + "\n");
	}
}
